package triDataStruct;

import java.util.concurrent.*;
import java.util.concurrent.locks.*;

import transaction.Transaction;

public abstract class MemoryBuffer {
	
	protected BufferObject[] buffer;	
	protected int[] pinCnt;
	protected boolean[] LRU;
	
	protected ConcurrentHashMap<Integer, Integer> map;
	protected ReentrantLock indexLock;

	public MemoryBuffer(int bufferSize) {
		buffer = new BufferObject[bufferSize];
		LRU = new boolean[bufferSize];
		pinCnt = new int[bufferSize];
		
		map = new ConcurrentHashMap<Integer, Integer>();
		indexLock = new ReentrantLock();
	}
	
	protected int get(int key, Object arg, Transaction tx) {
		Integer index = map.get(key);
		
		if (index != null) {
			synchronized(buffer[index]) {
				if (buffer[index].getKey() == key) {
					++pinCnt[index];
					LRU[index] = true;
					return index;
				}
			}
		}
		
		index = loadInBuffer(key, arg, tx);
		
		return index;
	}
	
	private int cursor = -1;
	
	//load into the buffer
	protected int loadInBuffer(int key, Object arg, Transaction tx) {
		indexLock.lock();
		try {
			Integer index = map.get(key);
			if (index != null) {
				synchronized(buffer[index]) {
					++pinCnt[index];
					LRU[index] = true;
					return index;
				}
			}
			
			//spinning the cursor
			while (true) {
				cursor = (cursor + 1) % buffer.length;
				
				if (buffer[cursor] == null) {
					buffer[cursor] = createEmptyObject(cursor, tx);
					
					this.loadObject(key, arg, cursor, tx);
					pinCnt[cursor] = 1;
					LRU[cursor] = true;
					map.put(key, cursor);
					
					//System.out.println("swap in " + inodePageId);
					return cursor;
				}
				
				synchronized(buffer[cursor]) {
					if (pinCnt[cursor] == 0 && !LRU[cursor] && buffer[cursor].tryRLock(tx)) {
						//System.out.println("swap out " + buffer[cursor].getInodePageId() + " with in " + inodePageId);
						
						map.remove(buffer[cursor].getKey());
						this.flushObject(cursor, tx);
						this.loadObject(key, arg, cursor, tx);
						pinCnt[cursor] = 1;
						LRU[cursor] = true;
						map.put(key, cursor);
						
						buffer[cursor].lock.unlockRead(tx);
						
						return cursor;
					}
					else if (pinCnt[cursor] == 0 && LRU[cursor]) {
						LRU[cursor] = false;
					}
				}
				
			}
		}
		finally {
			indexLock.unlock();	
		}
	}
	
	protected void unPin(int index) {
		synchronized (buffer[index]) {
			--pinCnt[index];
		}
	}
	
	protected abstract BufferObject createEmptyObject(int index, Transaction tx);
	
	protected abstract void loadObject(int key, Object arg, int index, Transaction tx);
	
	protected abstract void flushObject(int index, Transaction tx);
}
